package project

import dmputils.PeiZhiFile
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/03/28.
  */
object DiyuFenBu2 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    .setMaster("local[*]")
    .setAppName("地域分布分析")
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc: SparkContext = new SparkContext(conf)
    val sqlc: SQLContext = new SQLContext(sc)
    val dataFrame = sqlc.read.parquet(PeiZhiFile.config.getString("inputpath"))

    val baseData: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double)] = dataFrame.map(row => {
      //省份名称
      val provincename: String = row.getAs[String]("provincename")
      //城市名称
      val cityname: String = row.getAs[String]("cityname")
      //数据请求方式（1请求、2展示、3点击）
      val requestmode: Int = row.getAs[Int]("requestmode")
      //流程节点（1：请求量 kpi 2：有效请求 3：广告请求
      val processnode: Int = row.getAs[Int]("processnode")
      //有效标识（有效指可以正常计费的）(0：无效 1：有效
      val iseffective: Int = row.getAs[Int]("iseffective")
      //是否收费（0：未收费 1：已收费）
      val isbilling: Int = row.getAs[Int]("isbilling")
      //是否 rtb
      val isbid: Int = row.getAs[Int]("isbid")
      //是否竞价成功
      val iswin: Int = row.getAs[Int]("iswin")
      // 广告 id
      val adorderid: Int = row.getAs[Int]("adorderid")
      //rtb 竞价成功价格 每次消费
      val winprice: Double = row.getAs[Double]("winprice")
      //转换后的广告消费
      val adpayment: Double = row.getAs[Double]("adpayment")

      (provincename, cityname, requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment)
      //      1         2           3             4           5         6           7       8       9         10         11
    })
    //baseData.collect().foreach(println)
    val pdData: RDD[(String, String, Int, Int, Int, Int, Int, Int, Int, Double, Double)] = baseData.map(t => {
      //原始请求数
      val Ysrequest = if (t._3 == 1 && t._4 >= 1) 1 else 0
      //有效请求数
      var Yxrequest = if (t._3 == 1 && t._4 >= 2) 1 else 0
      //广告请求数
      var Ggrequest = if (t._3 == 1 && t._4 == 3) 1 else 0
      //参与竞价数
      var Cyjinjia = if (t._5 == 1 && t._6 == 1 && t._7 == 1 && t._9 != 0) 1 else 0
      //竞价成功数
      var Jjsuccess = if (t._5 == 1 && t._6 == 1 && t._8 == 1) 1 else 0
      //展示数
      var Zssome = if (t._3 == 2 && t._5 == 1) 1 else 0
      //点击数
      var Djsome = if (t._3 == 3 && t._5 == 1) 1 else 0
      //DSP广告消费
      var Xfmoney = if (t._5 == 1 && t._6 == 1 && t._8 == 1) t._9/1000 else 0
      //DSP广告成本
      var Cbmoney = if (t._5 == 1 && t._6 == 1 && t._8 == 1) t._10/1000 else 0

      (t._1, t._2, Ysrequest, Yxrequest, Ggrequest, Cyjinjia, Jjsuccess, Zssome, Djsome, Xfmoney, Cbmoney)

    })
    //pdData.collect().foreach(println)
   //   println(pdData.count())

    val data1: RDD[((String, String), List[Double])] = pdData.map(t => {

      ((t._1, t._2), List(t._3, t._4, t._5, t._6, t._7, t._8, t._9, t._10, t._11))

    })


    val key: RDD[((String, String), List[Double])] = data1.reduceByKey((a, b) => {

      a.zip(b).map(value => value._1 + value._2)

    })
    key.saveAsTextFile(PeiZhiFile.config.getString("outputpath"))

    sc.stop()
  }

}
